package com.apps.sdses.flink141.app;

import com.alibaba.fastjson2.JSON;
import com.apps.sdses.flink141.source.ClickhouseSource;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.RowDataSerializer;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.DataType;

import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;


public class FlinkJarClickhouse2Doris {
    public static void main(String[] args) {
        // 创建Flink执行环境
//        Configuration jobConfiguration = new Configuration();
        //设置WebUI绑定的本地端口
        //jobConfiguration.setInteger(RestOptions.PORT, 18081);//"rest.port" / RestOptions.PORT 均可 | 注: 8081 是默认端口
//        jobConfiguration.setString(RestOptions.BIND_PORT, "18081");
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(jobConfiguration);
        env.enableCheckpointing(10000);

        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, Time.of(10, TimeUnit.SECONDS)));

        // 数据源，读clickhouse，并行度按需设置
        DataStream<Map<String, String>> sourceDS = env.addSource(new ClickhouseSource()).setParallelism(5);

//        sourceDS.print();

        SingleOutputStreamOperator<String> mapDS = sourceDS.map(new MapFunction<Map<String, String>, String>() {
            @Override
            public String map(Map<String, String> mapData) throws Exception {

                //TODO
                return JSON.toJSONString(mapData);
            }
        });
//        mapDS.print();


//        SingleOutputStreamOperator<RowData> mapDS = sourceDS.map(new MapFunction<Map<String, String>, RowData>() {
//            @Override
//            public RowData map(Map<String, String> mapData) throws Exception {
//                GenericRowData genericRowData = new GenericRowData(18);
//                genericRowData.setField(0, StringData.fromString(mapData.get("WDY_ID")));
//                genericRowData.setField(1, StringData.fromString(mapData.get("RealTime")));
//                genericRowData.setField(2, StringData.fromString(mapData.get("SaveTime")));
//                genericRowData.setField(3, StringData.fromString(mapData.get("TaskTime")));
//                genericRowData.setField(4, Long.valueOf(mapData.get("ID")));
//                genericRowData.setField(5, StringData.fromString(mapData.get("PD_ID")));
//                genericRowData.setField(6, StringData.fromString(mapData.get("MpNo")));
//                genericRowData.setField(7, StringData.fromString(mapData.get("EquipNo")));
//                genericRowData.setField(8, StringData.fromString(mapData.get("EquipTypeCode")));
//                genericRowData.setField(9, DecimalData.fromBigDecimal(, 18, 2)));
//                genericRowData.setField(10, Double.valueOf(mapData.get("WdyHUM")));
//                genericRowData.setField(11, StringData.fromString(mapData.get("AlarmStatus")));
//                genericRowData.setField(12, StringData.fromString(mapData.get("ErrorDetail")));
//                genericRowData.setField(13, StringData.fromString(mapData.get("ExtendField")));
//                genericRowData.setField(14, null);
//                genericRowData.setField(15, null);
//                genericRowData.setField(16, Long.valueOf(mapData.get("Status")));
//                genericRowData.setField(17, null);
//                return genericRowData;
//            }
//        });

//        mapDS.print();
//
        //doris sink option
        DorisSink.Builder<String> builder = DorisSink.builder();
        DorisOptions.Builder dorisBuilder = DorisOptions.builder();
        dorisBuilder.setFenodes("172.21.32.230:38030,172.21.32.231:38030")
                .setTableIdentifier("db_datacube_doris_test.t_data_wdy_real_history")
                .setUsername("datacube_doris_test")
                .setPassword("Nx8Tz_Y8v3x");

        // json format to streamload
        Properties properties = new Properties();
        properties.setProperty("format", "json");
        properties.setProperty("read_json_by_line", "true");
        DorisExecutionOptions.Builder  executionBuilder = DorisExecutionOptions.builder();
        executionBuilder.setLabelPrefix("label-doris20241119") //streamload label prefix
                .setDeletable(false)
                .setStreamLoadProp(properties); //streamload params
//
//        //flink rowdata‘s schema
//        String[] fields = {"wdy_id"," real_time"," save_time"," task_time"," id"," pd_id"," mpno"," equip_no"," equip_type_code"," wdy_temp"," wdy_hum"," alarm_status"," error_detail"," extend_field"," tz_wdy_temp"," temperature_deviation"," status", "abnormal_data"};
//        DataType[] types = {
//                DataTypes.VARCHAR(64),
//                DataTypes.STRING(),
//                DataTypes.STRING(),
//                DataTypes.STRING(),
//                DataTypes.BIGINT(),
//                DataTypes.VARCHAR(64),
//                DataTypes.VARCHAR(64),
//                DataTypes.VARCHAR(64),
//                DataTypes.VARCHAR(64),
//                DataTypes.DECIMAL(18, 2),
//                DataTypes.DECIMAL(18, 2),
//                DataTypes.VARCHAR(64), DataTypes.STRING(), DataTypes.STRING(),
//                DataTypes.DECIMAL(18, 2),
//                DataTypes.DECIMAL(18, 2),
//                DataTypes.INT(),
//                DataTypes.STRING()};

        builder.setDorisReadOptions(DorisReadOptions.builder().build())
                .setDorisExecutionOptions(executionBuilder.build())
                .setSerializer(new SimpleStringSerializer())
                .setDorisOptions(dorisBuilder.build());

        mapDS.sinkTo(builder.build());

        // execute
        try {
            env.execute();
        } catch (Exception e) {
            System.out.println(e.getMessage());
        }
    }
}
